package simpledb.optimizer;

import simpledb.ParsingException;
import simpledb.common.Database;
import simpledb.common.Type;
import simpledb.execution.*;
import simpledb.storage.*;
import simpledb.transaction.TransactionId;

import java.io.File;
import java.util.*;

/**
 * LogicalPlan 表示一个已经通过解析器处理、准备被优化器处理的逻辑查询计划。
 *
 * 它包含一系列表扫描节点（scan）、连接节点（join）、过滤条件（filter），
 * 以及 select 列表和 group by 字段等信息。
 *
 * LogicalPlan 只能表示具有一个聚合字段和一个分组字段的查询。
 *
 * 可以使用 {@link #physicalPlan} 方法将逻辑计划转换为物理执行计划（经过优化的计划），
 * 该方法会调用 JoinOptimizer 来对连接顺序进行优化，并选择最佳的连接实现方式。
 */
public class LogicalPlan {

    // 所有连接操作的节点列表
    private List<LogicalJoinNode>             joins;

    // 所有表扫描节点（每个表对应一个 LogicalScanNode）
    private final List<LogicalScanNode>       tables;

    // 所有过滤条件节点
    private final List<LogicalFilterNode>     filters;

    // 子查询映射表，用于保存子查询对应的迭代器
    private final Map<String, OpIterator>     subplanMap;

    // 表别名到表ID的映射
    private final Map<String, Integer>        tableMap;

    // SELECT 子句中的字段列表（含聚合函数）
    private final List<LogicalSelectListNode> selectList;

    // GROUP BY 的字段名称
    private String                            groupByField = null;

    // 是否包含聚合操作
    private boolean                           hasAgg       = false;

    // 聚合操作符，如 COUNT、SUM 等
    private String                            aggOp;

    // 被聚合的字段名
    private String                            aggField;

    // 排序方向：升序还是降序
    private boolean                           oByAsc, hasOrderBy = false;

    // 排序字段名
    private String                            oByField;

    // 查询语句字符串
    private String                            query;

    /** 构造函数 -- 创建一个空的逻辑查询计划 */
    public LogicalPlan() {
        joins = new ArrayList<>();
        filters = new ArrayList<>();
        tables = new ArrayList<>();
        subplanMap = new HashMap<>();
        tableMap = new HashMap<>();

        selectList = new ArrayList<>();
        this.query = "";
    }

    /**
     * 设置当前逻辑计划所代表的 SQL 查询文本。
     * 注意：不会解析该查询文本，只是用来记录生成该计划的原始 SQL。
     *
     * @param query 与该计划关联的 SQL 查询文本
     */
    public void setQuery(String query) {
        this.query = query;
    }

    /**
     * 获取通过 setQuery 设置的查询文本。
     *
     * @return 当前逻辑计划的 SQL 查询文本
     */
    public String getQuery() {
        return query;
    }

    /**
     * 根据表别名获取其对应的数据库表 ID。
     * 别名是通过 addScan 添加的表的别名。
     *
     * @param alias 表的别名
     * @return 对应表的 ID，若不存在则返回 null
     */
    public Integer getTableId(String alias) {
        return tableMap.get(alias);
    }

    /**
     * 获取表别名到表 ID 的映射关系。
     *
     * @return 表别名到表 ID 的映射表
     */
    public Map<String, Integer> getTableAliasToIdMapping() {
        return this.tableMap;
    }

    /**
     * 向逻辑计划中添加一个新的过滤条件。
     *
     * @param field 过滤字段名（可以是全限定名，也可以是唯一字段名）
     * @param p 过滤谓词（操作符）
     * @param constantValue 与字段比较的常量值
     * @throws ParsingException 如果字段不在已添加的表中或存在歧义
     */
    public void addFilter(String field, Predicate.Op p, String constantValue) throws ParsingException {

        field = disambiguateName(field); // 消除字段名歧义
        String table = field.split("[.]")[0]; // 提取表名

        LogicalFilterNode lf = new LogicalFilterNode(table, field.split("[.]")[1], p, constantValue);
        filters.add(lf);
    }

    /**
     * 添加两个字段之间的连接操作。
     *
     * @param joinField1 第一个连接字段（可全限定）
     * @param joinField2 第二个连接字段
     * @param pred 连接谓词
     * @throws ParsingException 如果字段存在歧义或不在任何表中
     */
    public void addJoin(String joinField1, String joinField2, Predicate.Op pred) throws ParsingException {
        joinField1 = disambiguateName(joinField1);
        joinField2 = disambiguateName(joinField2);

        String table1Alias = joinField1.split("[.]")[0];
        String table2Alias = joinField2.split("[.]")[0];

        if (table1Alias.equals(table2Alias))
            throw new ParsingException("不能在同一个表的两个字段之间建立连接");

        String pureField1 = joinField1.split("[.]")[1];
        String pureField2 = joinField2.split("[.]")[1];

        LogicalJoinNode lj = new LogicalJoinNode(table1Alias, table2Alias, pureField1, pureField2, pred);
        System.out.println("Added join between " + joinField1 + " and " + joinField2);
        joins.add(lj);
    }

    /**
     * 添加一个字段与子查询之间的连接操作。
     *
     * @param joinField1 第一个连接字段（可全限定）
     * @param joinField2 子查询，连接字段为其结果集的第一列
     * @param pred 连接谓词
     * @throws ParsingException 如果字段存在歧义或不在任何表中
     */
    public void addJoin(String joinField1, OpIterator joinField2, Predicate.Op pred) throws ParsingException {
        joinField1 = disambiguateName(joinField1);

        String table1 = joinField1.split("[.]")[0];
        String pureField = joinField1.split("[.]")[1];

        LogicalSubplanJoinNode lj = new LogicalSubplanJoinNode(table1, pureField, joinField2, pred);
        System.out.println("Added subplan join on " + joinField1);
        joins.add(lj);
    }

    /**
     * 向计划中添加一个表扫描操作。每个表的每个别名都需要添加一次。
     *
     * @param table 表的 ID（可通过 Catalog 获取 DbFile）
     * @param name 表在计划中的别名
     */
    public void addScan(int table, String name) {
        System.out.println("Added scan of table " + name);
        tables.add(new LogicalScanNode(table, name));
        tableMap.put(name, table);
    }

    /**
     * 向 SELECT 列表中添加指定字段及其聚合操作。
     * 输出字段顺序按照添加顺序反向排列（即最后添加的字段最先输出）。
     *
     * @param fname 要投影的字段名
     * @param aggOp 聚合操作符
     * @throws ParsingException 如果字段无法识别或存在歧义
     */
    public void addProjectField(String fname, String aggOp) throws ParsingException {
        fname = disambiguateName(fname);
        if (fname.equals("*"))
            fname = "null.*";
        System.out.println("Added select list field " + fname);
        if (aggOp != null) {
            System.out.println("\t with aggregator " + aggOp);
        }
        selectList.add(new LogicalSelectListNode(aggOp, fname));
    }

    /**
     * 向查询中添加聚合操作和分组字段。
     * SimpleDB 仅支持单个聚合表达式和一个 GROUP BY 字段。
     *
     * @param op 聚合操作符（如 SUM、COUNT 等）
     * @param afield 要聚合的字段
     * @param gfield 分组字段
     * @throws ParsingException 如果字段无法识别或存在歧义
     */
    public void addAggregate(String op, String afield, String gfield) throws ParsingException {
        afield = disambiguateName(afield);
        if (gfield != null)
            gfield = disambiguateName(gfield);
        aggOp = op;
        aggField = afield;
        groupByField = gfield;
        hasAgg = true;
    }

    /**
     * 添加一个 ORDER BY 表达式，按照指定字段和排序顺序进行排序。
     * SimpleDB 目前只支持单个 ORDER BY 字段。
     *
     * @param field 要排序的字段名（可以是全限定名）
     * @param asc 排序方向：true 表示升序，false 表示降序
     * @throws ParsingException 如果字段无法识别或存在歧义
     */
    public void addOrderBy(String field, boolean asc) throws ParsingException {
        field = disambiguateName(field); // 消除字段名歧义，确定具体属于哪个表
        oByField = field; // 记录排序字段
        oByAsc = asc; // 记录排序方向
        hasOrderBy = true; // 标记当前查询包含 ORDER BY
    }

    /**
     * 给定一个字段名，尝试通过所有已添加的表来判断该字段属于哪个表。
     * 返回格式为 "tableAlias.fieldName" 的全限定名。
     *
     * 如果传入的字段已经是全限定名，则直接返回；
     * 如果字段不存在于任何表中，抛出 ParsingException；
     * 如果字段在多个表中都存在，也抛出 ParsingException。
     *
     * @param name 要解析的字段名
     * @return 全限定字段名（如 tableAlias.fieldName）
     * @throws ParsingException 如果字段不存在或存在歧义
     */
    String disambiguateName(String name) throws ParsingException {
        String[] fields = name.split("[.]");

        // 如果已经是 table.field 形式且不是 null.field，直接返回原值
        if (fields.length == 2 && (!fields[0].equals("null")))
            return name;

        // 如果有超过两个点号，非法字段名
        if (fields.length > 2)
            throw new ParsingException("Field " + name + " is not a valid field reference.");

        // 如果是 null.field 或 "*"，去掉 null 前缀继续查找
        if (fields.length == 2)
            name = fields[1];
        if (name.equals("*"))
            return name;

        // 遍历所有扫描过的表，查找字段是否存在于某个表中
        Iterator<LogicalScanNode> tableIt = tables.iterator();
        String tableName = null;
        while (tableIt.hasNext()) {
            LogicalScanNode table = tableIt.next();
            try {
                TupleDesc td = Database.getCatalog().getDatabaseFile(table.t).getTupleDesc();

                // 如果该表中存在这个字段
                td.fieldNameToIndex(name);
                if (tableName == null) {
                    tableName = table.alias; // 第一次找到记录表名
                } else {
                    // 多个表中都有这个字段，存在歧义，抛异常
                    throw new ParsingException(
                        "Field " + name + " appears in multiple tables; disambiguate by referring to it as tablename."
                                + name);
                }
            } catch (NoSuchElementException e) {
                // 忽略不存在该字段的表
            }
        }

        // 找到了唯一的匹配表
        if (tableName != null)
            return tableName + "." + name;
        else
            throw new ParsingException("Field " + name + " does not appear in any tables.");
    }

    /**
     * 将聚合操作符字符串转换为 Aggregator.Op 枚举类型。
     *
     * @param s 聚合操作符名称（如 "SUM", "COUNT" 等）
     * @return 对应的 Aggregator.Op 枚举值
     * @throws ParsingException 如果传入的字符串不是合法的聚合操作符
     */
    static Aggregator.Op getAggOp(String s) throws ParsingException {
        s = s.toUpperCase(); // 转换为大写统一比较
        if (s.equals("AVG"))
            return Aggregator.Op.AVG;
        if (s.equals("SUM"))
            return Aggregator.Op.SUM;
        if (s.equals("COUNT"))
            return Aggregator.Op.COUNT;
        if (s.equals("MIN"))
            return Aggregator.Op.MIN;
        if (s.equals("MAX"))
            return Aggregator.Op.MAX;
        throw new ParsingException("Unknown predicate " + s);
    }

    /**
     * 将当前逻辑计划（LogicalPlan）转换为一个物理计划（Physical Plan），返回一个 OpIterator。
     * 通过使用 JoinOptimizer.orderJoins 方法对连接顺序进行优化，尝试找到最优执行计划。
     *
     * @param t 当前事务对象，用于执行此 OpIterator
     * @param baseTableStats 提供每个表的统计信息（如行数、字段分布等）
     *                       键为表名（非别名），值为 TableStats 对象
     * @param explain 是否输出可视化物理执行计划
     * @throws ParsingException 如果逻辑计划不合法或解析失败
     * @return 返回表示该查询的物理执行计划的 OpIterator
     */
    public OpIterator physicalPlan(TransactionId t, Map<String, TableStats> baseTableStats, boolean explain)
                                                                                                            throws ParsingException {
        // 遍历所有逻辑扫描节点（即 FROM 子句中的表）
        Iterator<LogicalScanNode> tableIt = tables.iterator();

        // 用于记录等价表别名，例如 join 后合并两个表为一个子计划后，t2 被认为等价于 t1
        Map<String, String> equivMap = new HashMap<>();

        // 每个表的过滤选择率（selectivity）：用于优化器估算代价
        Map<String, Double> filterSelectivities = new HashMap<>();

        // 每个表的统计信息映射
        Map<String, TableStats> statsMap = new HashMap<>();

        // 第一步：创建 SeqScan 算子（顺序扫描）
        while (tableIt.hasNext()) {
            LogicalScanNode table = tableIt.next();
            SeqScan ss = null;
            try {
                // 获取实际数据库文件 ID 并创建 SeqScan
                ss = new SeqScan(t, Database.getCatalog().getDatabaseFile(table.t).getId(), table.alias);
            } catch (NoSuchElementException e) {
                throw new ParsingException("未知表 " + table.t);
            }

            // 记录每个表别名对应的子计划（初始为 SeqScan）
            subplanMap.put(table.alias, ss);

            // 获取基础表名（去除别名）
            String baseTableName = Database.getCatalog().getTableName(table.t);

            // 统计信息和选择率初始化
            statsMap.put(baseTableName, baseTableStats.get(baseTableName));
            filterSelectivities.put(table.alias, 1.0); // 初始选择率为 1（未过滤）
        }

        // 第二步：处理 WHERE 子句中的 Filter，将 Filter 算子封装在 SeqScan 上
        for (LogicalFilterNode lf : filters) {
            OpIterator subplan = subplanMap.get(lf.tableAlias);
            if (subplan == null) {
                throw new ParsingException("WHERE 条件中引用了未知表 " + lf.tableAlias);
            }

            Field f;
            Type ftyp;
            TupleDesc td = subplan.getTupleDesc();

            try {
                ftyp = td.getFieldType(td.fieldNameToIndex(lf.fieldQuantifiedName));
            } catch (NoSuchElementException e) {
                throw new ParsingException("Filter 表达式中引用了未知字段 " + lf.fieldQuantifiedName);
            }

            // 根据字段类型构造常量字段
            if (ftyp == Type.INT_TYPE)
                f = new IntField(new Integer(lf.c));
            else
                f = new StringField(lf.c, Type.STRING_LEN);

            Predicate p = null;
            try {
                // 构造谓词条件
                p = new Predicate(subplan.getTupleDesc().fieldNameToIndex(lf.fieldQuantifiedName), lf.p, f);
            } catch (NoSuchElementException e) {
                throw new ParsingException("未知字段 " + lf.fieldQuantifiedName);
            }

            // 替换原来的 SeqScan 为 Filter 算子
            subplanMap.put(lf.tableAlias, new Filter(p, subplan));

            // 更新该表的选择率（selectivity）
            TableStats s = statsMap.get(Database.getCatalog().getTableName(this.getTableId(lf.tableAlias)));
            double sel = s
                .estimateSelectivity(subplan.getTupleDesc().fieldNameToIndex(lf.fieldQuantifiedName), lf.p, f);
            filterSelectivities.put(lf.tableAlias, filterSelectivities.get(lf.tableAlias) * sel);
        }

        // 第三步：使用 JoinOptimizer 优化连接顺序
        JoinOptimizer jo = new JoinOptimizer(this, joins);
        joins = jo.orderJoins(statsMap, filterSelectivities, explain); // 得到优化后的连接顺序

        // 第四步：根据优化后的连接顺序，生成 Join 算子
        for (LogicalJoinNode lj : joins) {
            OpIterator plan1, plan2;
            boolean isSubqueryJoin = lj instanceof LogicalSubplanJoinNode;

            String t1name, t2name;

            // 处理等价表别名（join 合并后可能 t2 已被归入 t1 的等价组）
            if (equivMap.get(lj.t1Alias) != null)
                t1name = equivMap.get(lj.t1Alias);
            else
                t1name = lj.t1Alias;

            if (equivMap.get(lj.t2Alias) != null)
                t2name = equivMap.get(lj.t2Alias);
            else
                t2name = lj.t2Alias;

            plan1 = subplanMap.get(t1name);

            if (isSubqueryJoin) {
                // 子查询连接
                plan2 = ((LogicalSubplanJoinNode) lj).subPlan;
                if (plan2 == null)
                    throw new ParsingException("无效的子查询。");
            } else {
                plan2 = subplanMap.get(t2name);
            }

            if (plan1 == null)
                throw new ParsingException("WHERE 条件中引用了未知表 " + lj.t1Alias);
            if (plan2 == null)
                throw new ParsingException("WHERE 条件中引用了未知表 " + lj.t2Alias);

            // 实例化 Join 算子，并更新 subplanMap
            OpIterator j = JoinOptimizer.instantiateJoin(lj, plan1, plan2);
            subplanMap.put(t1name, j);

            if (!isSubqueryJoin) {
                // 不是子查询，则删除 t2 的子计划，并标记其等价于 t1
                subplanMap.remove(t2name);
                equivMap.put(t2name, t1name);

                // 更新其他指向 t2 的等价关系为 t1
                for (Map.Entry<String, String> s : equivMap.entrySet()) {
                    if (s.getValue().equals(t2name)) {
                        s.setValue(t1name);
                    }
                }
            }
        }

        // 所有连接完成后，应只剩下一个主计划
        if (subplanMap.size() > 1) {
            throw new ParsingException("查询没有包含连接所有表的 JOIN 条件！");
        }

        // 取出最终的执行计划根节点
        OpIterator node = subplanMap.entrySet().iterator().next().getValue();

        // 第五步：处理 SELECT 子句，构造投影字段列表和类型
        List<Integer> outFields = new ArrayList<>();
        List<Type> outTypes = new ArrayList<>();

        for (int i = 0; i < selectList.size(); i++) {
            LogicalSelectListNode si = selectList.get(i);

            if (si.aggOp != null) {
                // 如果是聚合函数
                outFields.add(groupByField != null ? 1 : 0);
                TupleDesc td = node.getTupleDesc();
                try {
                    td.fieldNameToIndex(si.fname); // 检查字段是否存在
                } catch (NoSuchElementException e) {
                    throw new ParsingException("SELECT 列表中存在未知字段 " + si.fname);
                }
                outTypes.add(Type.INT_TYPE); // 聚合结果默认为 INT 类型
            } else if (hasAgg) {
                // 如果有聚合但不是聚合字段，则必须出现在 GROUP BY 中
                if (groupByField == null) {
                    throw new ParsingException("字段 " + si.fname + " 未出现在 GROUP BY 列表中");
                }
                outFields.add(0);
                TupleDesc td = node.getTupleDesc();
                int id;
                try {
                    id = td.fieldNameToIndex(groupByField);
                } catch (NoSuchElementException e) {
                    throw new ParsingException("GROUP BY 字段 " + groupByField + " 不存在");
                }
                outTypes.add(td.getFieldType(id));
            } else if (si.fname.equals("null.*")) {
                // SELECT * 的情况，投影所有字段
                TupleDesc td = node.getTupleDesc();
                for (i = 0; i < td.numFields(); i++) {
                    outFields.add(i);
                    outTypes.add(td.getFieldType(i));
                }
            } else {
                // 普通字段投影
                TupleDesc td = node.getTupleDesc();
                int id;
                try {
                    id = td.fieldNameToIndex(si.fname);
                } catch (NoSuchElementException e) {
                    throw new ParsingException("SELECT 列表中存在未知字段 " + si.fname);
                }
                outFields.add(id);
                outTypes.add(td.getFieldType(id));
            }
        }

        // 第六步：如果有聚合操作，添加 Aggregate 算子
        if (hasAgg) {
            TupleDesc td = node.getTupleDesc();
            Aggregate aggNode;
            try {
                aggNode = new Aggregate(node, td.fieldNameToIndex(aggField),
                    groupByField == null ? Aggregator.NO_GROUPING : td.fieldNameToIndex(groupByField), getAggOp(aggOp));
            } catch (NoSuchElementException | IllegalArgumentException e) {
                throw new simpledb.ParsingException(e);
            }
            node = aggNode;
        }

        // 第七步：如果有 ORDER BY，添加 OrderBy 算子
        if (hasOrderBy) {
            node = new OrderBy(node.getTupleDesc().fieldNameToIndex(oByField), oByAsc, node);
        }

        // 最终返回 Project 算子，按 SELECT 列表投影字段
        return new Project(outFields, outTypes, node);
    }

    public static void main(String[] argv) {
        // 构造一个包含 3 个字段的表结构（TupleDesc）
        Type[] types = new Type[] { Type.INT_TYPE, Type.INT_TYPE, Type.INT_TYPE };
        String[] names = new String[] { "field0", "field1", "field2" };

        TupleDesc td = new TupleDesc(types, names); // 表结构描述
        TableStats ts;
        Map<String, TableStats> tableMap = new HashMap<>(); // 存储表统计信息

        // 创建 HeapFile 类型的表文件，并注册到 Catalog 中
        HeapFile table1 = new HeapFile(new File("some_data_file1.dat"), td);
        Database.getCatalog().addTable(table1, "t1"); // 将表 t1 添加进数据库目录

        // 创建该表的统计信息对象，并加入 map
        ts = new TableStats(table1.getId(), 1);
        tableMap.put("t1", ts);

        // 创建事务 ID
        TransactionId tid = new TransactionId();

        // 创建逻辑计划对象
        LogicalPlan lp = new LogicalPlan();

        // 添加扫描节点：从表 t1 中读取数据
        lp.addScan(table1.getId(), "t1");

        try {
            // 添加过滤条件：WHERE t1.field0 > 1
            lp.addFilter("t1.field0", Predicate.Op.GREATER_THAN, "1");
        } catch (Exception ignored) {
            // 忽略异常（在实际中应做适当处理）
        }

        /*
         * 下面是手动构建执行计划的示例（被注释掉）：
         *
         * SeqScan ss1 = new SeqScan(tid, table1.getId(), "t1");
         * SeqScan ss2 = new SeqScan(tid, table2.getId(), "t2");
         *
         * Filter sf1 = new Filter(
         *     new Predicate(0, Predicate.Op.GREATER_THAN, new IntField(1)),
         *     ss1);
         *
         * JoinPredicate p = new JoinPredicate(1, Predicate.Op.EQUALS, 1);
         * Join j = new Join(p, sf1, ss2);
         */

        OpIterator j = null;
        try {
            // 根据逻辑计划生成物理执行计划
            j = lp.physicalPlan(tid, tableMap, false);
        } catch (ParsingException e) {
            e.printStackTrace();
            System.exit(0);
        }

        // 执行查询并输出结果
        try {
            j.open(); // 打开迭代器，准备读取数据
            while (j.hasNext()) {
                Tuple tup = j.next(); // 获取下一条记录
                System.out.println(tup); // 输出记录内容
            }
            j.close(); // 关闭迭代器

            // 提交事务
            Database.getBufferPool().transactionComplete(tid);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
